/*
 * Copyright (C) 2015 RGG. All rights reserved.
 */
package com.taoxeo.mqtt;

import java.util.ArrayList;
import java.util.List;

import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttSecurityException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The Class MQTTClient.
 *
 * @author Derek
 * @version 1.0
 * @date 2015-11-18
 */
public class MQTTClient {
	public final static String USER_PATH = "USER";
	public final static String USER_PATH_withSeparator = USER_PATH + "/";
	public final static String APP_PATH = "APP";
	public final static String APP_PATH_withSeparator = APP_PATH + "/";

	/** The mqtt client. */
	private IMqttAsyncClient mqttClient;

	/** The mqtt connect options. */
	private MqttConnectOptions mqttConnectOptions;

	/** The reconnect delay. */
	private int reconnectDelay = 5;

	/** The logger. */
	private Logger logger = LoggerFactory.getLogger("MQTT");

	/** The retry times. */
	private int retryTimes = 0;

	/** The message callback. */
	private MessageCallback messageCallback;

	/** The default qos. */
	private int defaultQos = 2;

	private List<String> destinations;

	private IMqttActionListener listener;

	private boolean retry = true;

	/**
	 * Instantiates a new MQTT client.
	 *
	 * @param serverURI the server uri
	 * @param options the options
	 */
	public MQTTClient(String serverURI, ConnectOptions options, final MessageCallback... messageCallback) {
		this(serverURI, MqttClient.generateClientId(), options, messageCallback);
	}

	/**
	 * Instantiates a new MQTT client.
	 *
	 * @param serverURI the server uri
	 * @param clientId the client id
	 * @param options the options
	 */
	public MQTTClient(String serverURI, String clientId, ConnectOptions options,
			final MessageCallback... messageCallback) {
		try {
			mqttClient = new MqttAsyncClient(serverURI, clientId, new MemoryPersistence());
			mqttConnectOptions = new MqttConnectOptions();
			mqttConnectOptions.setUserName(options.getUser());
			mqttConnectOptions.setPassword(options.getPassword().toCharArray());
			mqttConnectOptions.setCleanSession(options.isCleanSession());
			mqttConnectOptions.setKeepAliveInterval(options.getKeepAliveInterval());
			defaultQos = options.getQoS();
		}
		catch (MqttException e) {
			e.printStackTrace();
		}

		destinations = new ArrayList<String>(1);

		if (messageCallback != null && messageCallback.length > 0) {
			this.messageCallback = messageCallback[0];
		}

		mqttClient.setCallback(new MqttCallback() {
			@Override
			public void messageArrived(String topic, MqttMessage message) throws Exception {
				if (MQTTClient.this.messageCallback != null) {
					if (logger.isDebugEnabled()) {
						logger.debug("Message Arrived from Topic [{}], message length:{}", topic,
								message.getPayload().length);
					}
					MQTTClient.this.messageCallback.messageArrived(topic, message.getPayload());
				}
			}

			@Override
			public void deliveryComplete(IMqttDeliveryToken token) {
			}

			@Override
			public void connectionLost(Throwable cause) {
				// Connection reset, Connection refused, Network is unreachable
				reconnect(cause);
			}
		});

		listener = new IMqttActionListener() {

			@Override
			public void onSuccess(IMqttToken token) {
				logger.warn("###MQTT SERVER Connected! {}", mqttClient.getServerURI());
				if (messageCallback != null && mqttConnectOptions.isCleanSession() && !destinations.isEmpty()) {
					for (String topic : destinations) {
						try {
							logger.debug("subscribe topic: {}", topic);
							mqttClient.subscribe(topic, defaultQos);
						}
						catch (MqttException e) {
							e.printStackTrace();
						}

					}
				}
			}

			@Override
			public void onFailure(IMqttToken token, Throwable ex) {
				if (ex instanceof MqttSecurityException) {
					logger.error("权限认证失败放弃连接");
					try {
						mqttClient.disconnect();
					}
					catch (MqttException e) {
						e.printStackTrace();
					}
				}
				else if (ex instanceof MqttException) {
					reconnect(ex);
				}
			}
		};
	}

	/**
	 * Connect.
	 */
	public void connect() {
		retry = true;
		try {
			if (!mqttClient.isConnected()) {
				logger.debug("{} Connecting...", mqttClient.getClientId());
				mqttClient.connect(mqttConnectOptions, "", listener);
			}
		}
		catch (MqttSecurityException e) {
			e.printStackTrace();
		}
		catch (MqttException e) {
			e.printStackTrace();
		}
	}

	private void reconnect(Throwable ex) {
		if (!retry) {
			logger.info("Manual disconnect, Give up reconnect!");
			return;
		}
		logger.info("Connect exception:{}. After {} seconds to reconnect! Retry times {}", ex.toString(),
				reconnectDelay, ++retryTimes);
		try {
			Thread.currentThread().sleep(reconnectDelay * 1000);
			connect();
		}
		catch (InterruptedException e1) {
			e1.printStackTrace();
		}
	}

	/**
	 * Disconnect.
	 */
	public void disconnect() {
		try {
			retry = false;
			logger.info("Disconnect...");
			mqttClient.disconnectForcibly();
			logger.info("Disconnected");
		}
		catch (MqttException e) {
			e.printStackTrace();
		}
	}

	/**
	 * Subscribe.
	 *
	 * @param messageCallback the message callback
	 */
	public void subscribe(String topic) {
		try {
			mqttClient.subscribe(topic, defaultQos);
			destinations.add(topic);
			logger.debug("subscribe topic: {}", topic);
		}
		catch (MqttException e) {
			logger.error("subscribe error: {}", e.getMessage());
		}
	}

	public void unsubscribe(String topic) {
		try {
			mqttClient.unsubscribe(topic);
			destinations.remove(topic);
		}
		catch (MqttException e) {
			logger.error("unsubscribe error: {}", e.getMessage());
		}
	}

	public void publish(String destination, byte[] payload, int qos, boolean retained) {
		try {
			mqttClient.publish(destination, payload, qos, retained);
		}
		catch (MqttException e) {
			logger.error("publish error: {}", e.getMessage());
		}
	}

	public void publishToUser(String user, byte[] payload, boolean retained) {
		publish(USER_PATH_withSeparator + user, payload, defaultQos, retained);
	}

	public void publishToApp(String appId, byte[] payload, boolean retained) {
		publish(APP_PATH_withSeparator + appId, payload, defaultQos, retained);
	}

	public void setReconnectDelay(int reconnectDelay) {
		this.reconnectDelay = reconnectDelay;
	}

	public void addSubscribeTopic(String destination) {
		destinations.add(destination);
	}

	public boolean isConnected() {
		return mqttClient.isConnected();
	}

	public IMqttAsyncClient getMqttClient() {
		return mqttClient;
	}

}
